package com.cube.reactive.guide.reactive;

import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @author cube.li
 * @date 2021/11/14 13:46
 * @description java9 ReactiveStream
 */
@Slf4j
public class ReactiveStream {

    public static void main(String[] args) throws InterruptedException {
        //创建一个发布者
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        //创建一个订阅者
        Flow.Subscriber subscriber = new Flow.Subscriber() {
            Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                log.info("建立订阅关系");
                this.subscription = subscription;
                subscription.request(1);
            }

            @Override
            public void onNext(Object o) {
                log.info("接收数据:{}", o);
                //业务处理,背压
                subscription.request(10);
            }

            @Override
            public void onError(Throwable throwable) {
                log.error("发生错误了");
            }

            @Override
            public void onComplete() {
                log.info("请求结束");
            }
        };

        //建立订阅关系
        publisher.subscribe(subscriber);
        //发送数据
        publisher.submit("hello world!");

        //异步处理,需要主线程阻塞才能看到日志
        Thread.sleep(1100000);
    }
}
